package com.gz.mr;

import com.gz.entity.PageCount;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import java.io.IOException;
import java.util.HashSet;

public class MrPV {
    public static void main(String[] args) throws Exception {
        
        
        
        Configuration conf = new Configuration();
//        conf.set("fs.defaultFS","hdfs://gz001:8020");
        Job job=Job.getInstance(conf);
        
        job.setJobName("MysqlDB");
        job.setJarByClass(MrPV.class);
        job.setMapperClass(MrPV.MrPVMapper.class);
        job.setReducerClass(MrPV.MrPVReducer.class);
        
        //配置数据库信息
        String driveclass="com.mysql.jdbc.Driver";
        String url="jdbc:mysql://localhost:3306/bdyunpan";
        String username="root";
        String password="123456";
        DBConfiguration.configureDB(job.getConfiguration(),driveclass,url,username,password);
        
        //设置数据库输入
        //需要通过总的记录数来计算切片
//        DBInputFormat.setInput(job,MysqlDBWritable.class,"select name,age from persion","select count(*) from persion");
    
        TextInputFormat.setInputPaths(job, args[0]);
//        TextOutputFormat.setOutputPath("");
        
        //设置数据库输出
        DBOutputFormat.setOutput(job,"page_count", "site_id", "pv", "uv", "ip", "new_user", "date_time");
//        DBInputFormat.setInput();
    
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(DBOutputFormat.class);
    
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        job.setOutputKeyClass(PageCount.class);
        job.setOutputValueClass(NullWritable.class);
        
        job.waitForCompletion(true);
    

        
        
        
    }
    
    public static class MrPVMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            
            // 过滤掉其他的请求
            if(!value.toString().contains("_utm.gif")) return;
            
            String[] vs = value.toString().split("####");
            
            if(vs.length != 10) return;
            
            // key是固定的，就是 pv
            // "pv" => 1
            // "pv" => 1
            // "pv" => 1
            // "pv" => 1
            // "pv" => 1
            // "pv" => 1
            // "pv" => 1
            // "pv" => 1
            context.write(new Text("pv"), new IntWritable(1));
            
        }
    }
    
    public static class MrPVReducer extends Reducer<Text,IntWritable,PageCount,NullWritable>{
    
//        values => {1, 1, 1, 1, 1, 1, 1, 1, 1}
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
    
//            HashSet<String> set = new HashSet();
//            set.add("192.168.1.1");
//            set.add("192.168.1.2");
//            set.add("192.168.1.3");
//            set.add("192.168.1.4");
//
//            HashSet<String> jin = new HashSet();
//            set.add("192.168.1.1");
//            set.add("192.168.1.2");
//
//            for(String s: jin){
//                if(!set.contains(s)){
//                    context.write(s, s);
//                }
//            }
            
            
            PageCount pageCount=new PageCount();
            int pv = 0;
            for(IntWritable it : values){
                pv +=it.get();
            }
            
            pageCount.setPv(pv);
            pageCount.setDateTime("2020-10-20");
    
            // 输出的 key 的数据类型 必须是 这个对象 pageCount
            context.write(pageCount, NullWritable.get());
    
        }
    }
    
}
